package com.lagou;

import com.alibaba.fastjson.JSON;
import com.lagou.codec.RpcEncoder;
import com.lagou.handler.UserServerHandler;
import com.lagou.codec.JSONSerializer;
import com.lagou.codec.RpcDecoder;
import com.lagou.protocol.NodeInfo;
import com.lagou.protocol.RpcRequest;
import com.lagou.protocol.RpcResponse;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.nio.charset.StandardCharsets;

@SpringBootApplication
public class RpcProviderApplication implements ApplicationRunner {


    public static void main(String[] args) {
        SpringApplication.run(RpcProviderApplication.class, args);
    }
    private CuratorFramework curatorClient;
    private String path;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        String ip = "127.0.0.1";
        int port = 8890;
        connectZk(ip, port);
        startServer(ip, port);
    }

    private void connectZk(String ip, int port) throws Exception {
        String zookeeperAddress = "192.168.200.200:2181";

        // 启动zookeeper
        curatorClient =
                CuratorFrameworkFactory.builder().
                        connectString(zookeeperAddress) //连接地址
                        .connectionTimeoutMs(2000) //连接超时时间
                        .sessionTimeoutMs(10000) //会话超时时间
                        .retryPolicy(new ExponentialBackoffRetry(1000, 3)) //重试策略
                        .namespace("rpc_manage") //命名空间,默认节点
                        .build();
        curatorClient.start();

        // 构造数据
        NodeInfo nodeInfo = new NodeInfo();
        nodeInfo.setIp(ip);
        nodeInfo.setPort(port);
        String content = JSON.toJSONString(nodeInfo);

        // 创建临时顺序节点
        path = curatorClient.
                create().
                creatingParentsIfNeeded().
                withMode(CreateMode.EPHEMERAL_SEQUENTIAL).
                forPath("/server_node", content.getBytes());
    }

    private void startServer(String ip, int port) throws InterruptedException {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();

        // 构造json编解码器
        final JSONSerializer jsonSerializer = new JSONSerializer();

        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup,workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new DelimiterBasedFrameDecoder(9126,
                                Unpooled.copiedBuffer("\n".getBytes(StandardCharsets.UTF_8))));
                        pipeline.addLast(new RpcDecoder<>(RpcRequest.class, jsonSerializer));
                        pipeline.addLast(new RpcEncoder(RpcResponse.class, jsonSerializer));
                        pipeline.addLast(new UserServerHandler(curatorClient, path));
                    }
                });
        System.out.println(String.format("start to bind %s:%s", ip, port));
        serverBootstrap.bind(ip,port).sync();
    }
}
